package cn.wapache.timed.zookeeper;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.net.URI;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.wapache.time.calendar.Calendar;
import cn.wapache.time.trigger.Trigger;
import cn.wapache.timed.Job;
import cn.wapache.timed.Scheduler;
import cn.wapache.timed.Task;
import cn.wapache.timed.serialization.Serializer;
import cn.wapache.timed.spi.SchedulerProvider;
import cn.wapache.timed.zookeeper.ZooKeepers.CreateStrategy;
import cn.wapache.timed.zookeeper.ZooKeepers.DeleteStrategy;
import cn.wapache.timed.zookeeper.ZooKeepers.SetDataStrategy;

/**
 * 
 * 基于zookeeper实现分布式定时任务调度器。
 * 
 * @author ykuang
 *
 */
public class ZooKeeperScheduler implements Scheduler, Watcher {

	private static Logger logger = LoggerFactory.getLogger(ZooKeeperScheduler.class);

	enum Status{
		STARTUP,RECOVERING,RUNNING,DISABLED,ERROR,SHUTDOWN;
	}

	private ZooKeepers zks = new ZooKeepers();
    private ZooKeeper zooKeeper;
	
    protected CountDownLatch countDownLatch = new CountDownLatch(1);

    private ZooKeeperScheduler.Options options;
	
    private ExecutorService executors;
    
    private SchedulerProvider provider;
	
    private Serializer serializer;
    
	/** 本实例的标识 */
    private String instanceId;
    
    private Status state;
    
    private boolean stop = false;
	
	// 轮询需要用到的一个变量
    private long minNext;
	
	public ZooKeeperScheduler(SchedulerProvider provider, Options options){
		this.options = Objects.requireNonNull(options, "argument <options> require non null");
		this.provider = Objects.requireNonNull(provider, "argument <provider> require non null");
		this.serializer = this.provider.getSerializer(this.options.getSerializer());
	}
	
    ///////////////////////////////////////////////////////////////////////////
    // ZooKeeper Connection and Watcher methods
    ///////////////////////////////////////////////////////////////////////////
    
    @Override
    public void process(WatchedEvent event) {

		if(logger.isTraceEnabled()){
			logger.trace("Scheduler #{} processing event={} ", options.getId(), event);
		}
    	
    	switch(event.getState()) {
        case SyncConnected:  
            // 连接建立成功,或者重连成功
        	countDownLatch.countDown();
        	logger.info("Scheduler #{} connected to ZooKeeper.", options.getId());
            break;
        case Expired:  
            // session过期,这是个非常严重的问题,有可能client端出现了问题,也有可能zk环境故障
        	logger.warn("Scheduler #{} ZooKeeper session expired.", options.getId());
            countDownLatch = new CountDownLatch(1);
        	try {
				connect();
			} catch (Exception e) {
				logger.error(e.getMessage(), e);
				stop();
			}
            break;  
        case Disconnected:  
        	// ZooKeeper客户端有自动重连功能
        	logger.warn("Scheduler #{} ZooKeeper connection broken.", options.getId());
            break;  
        case AuthFailed:
        	logger.warn("Scheduler #{} ZooKeeper auth failed.", options.getId()); 
        	countDownLatch.countDown();
        	stop();
        default:  
            break;  
    	}  
    }
    
	private void connect() throws IOException {

    	logger.info("Scheduler #{} connecting to ZooKeeper.", options.getId());

        ZooKeeper zk = new ZooKeeper(options.getUrl(), options.getTimeout(), this);
    	if(options.getUserName()!=null){
    		zk.addAuthInfo("digest", (options.getUserName()+":"+options.getPassword()).getBytes());        	
    	}
    	
        zooKeeper = zk;
        zks.setZooKeeper(zk);
	}

	///////////////////////////////////////////////////////////////////////////
	// operation method
	///////////////////////////////////////////////////////////////////////////


    ///////////////////////////////////////////////////////////////////////////
    // Scheduler methods
    ///////////////////////////////////////////////////////////////////////////

	@Override
	public void start(){
		if(state==null){
			run();
		}
	}
	
	@Override
	public void pause(){
		options.setEnabled(false);
	}

	@Override
	public void resume(){
		options.setEnabled(true);
	}

	@Override
	public void stop(){
		this.stop = true;
	}

	@Override
	public boolean isPaused() {
		return !options.isEnabled();
	}
	
    ///////////////////////////////////////////////////////////////////////////
    // Runnable methods
    ///////////////////////////////////////////////////////////////////////////
	
	@Override
	public void run() {
		try{

			logger.info("Scheduler #{} start", options.getId());

			// 连接
			connect();
			
			// 初始化
			setup();
			
			// 检查初始状态
			switch(state){
			case STARTUP:
			case RECOVERING:
			case DISABLED:
				break;
			case ERROR:
			case RUNNING:
			case SHUTDOWN:
			default:
				logger.error("Scheduler #{} start up failed, status = {}", options.getId(), state);
				return;
			}
			
			// 轮询
			while( !stop && !Thread.interrupted()){

				minNext = 0;
				
				// 本轮开始时间
				long roundStart = System.currentTimeMillis();
				
				switch(state){
				case STARTUP:
				case RECOVERING:
					updateState(options.isEnabled() ? Status.RUNNING : Status.DISABLED);
					break;
				case RUNNING:
					
					long time1 = System.currentTimeMillis();
					
					List<String> children = zks.getChildren(options.getTriggersPath());
					
					long time2 = System.currentTimeMillis();
					
					Collections.shuffle(children); // 打乱扫描顺序, 让task更均衡地分布在各个scheduler

					logger.debug("Scheduler #{} got {} triggers", options.getId(), children.size());
					if(logger.isTraceEnabled()){
						logger.trace("Scheduler #{} scanning triggers: {} ", options.getId(), children);
					}

					long time3 = System.currentTimeMillis();
					
					for(String triggerId : children){
						poll(triggerId);
					}

					long time4 = System.currentTimeMillis();
					
					if(logger.isDebugEnabled()){
						logger.debug("Scheduler #{} scan used time, getChildren: {} MS, shuffle: {} MS, poll:{}", 
								options.getId(), (time2-time1),(time3-time2),(time4-time3));						
					}
					
					break;
				case DISABLED:
					if(options.isEnabled()){
						updateState(Status.RUNNING);
					}
					break;
				case ERROR:
					if(logger.isTraceEnabled()){
						logger.trace("Scheduler #{} in status {}", options.getId(), state);
					}
					break;
				case SHUTDOWN:
					stop();
					break;
				}
				
				// 休眠
				long roundEnd = System.currentTimeMillis();
				long time = minNext - roundEnd;
				time = time<100 ? 100 : time>1000 ? 1000 : time; // 控制休眠时间范围
				time = time - (roundEnd - roundStart); // 扣除本次轮询耗时
				if(time>15){ // 时间差太小,则不休眠
					
					if(logger.isTraceEnabled()){
						logger.trace("Scheduler #{} sleep {} MS", options.getId(), time);
					}
					
					Thread.sleep(time);
					
				}
			}
			
		} catch (Exception e) {
			logger.error("Scheduler #{} error, message={}",options.getId(),e.getMessage(),e);
		}finally{
			teardown(); // 释放资源

			logger.info("Scheduler #{} stop", options.getId());
		}
	}

    ///////////////////////////////////////////////////////////////////////////
    // private methods 
    ///////////////////////////////////////////////////////////////////////////
	
	private void setup() {
		
		instanceId = options.getId() + "#" + Integer.toHexString(System.identityHashCode(this)).toUpperCase();
		
		// 上锁
		state = Status.STARTUP;
		if(! zks.createEphemeralRecursively(options.getStatusPath(), state.toString().getBytes())){
			state = Status.ERROR;
			logger.info("Scheduler #{} setup error, create status znode failed, path={}", options.getId(), options.getStatusPath());
			return;
		}
		
		// 更新配置
		options = options.merge(zks.readObject(options.getOptionsPath(), serializer, options));
		if(null == zks.writeObject(options.getOptionsPath(), options, SetDataStrategy.CreateIfNotExists)){
			state = Status.ERROR;
			logger.info("Scheduler #{} error, message={}", options.getId(), "更新配置节点失败, 停止调度。");
			return;
		}
		
		// 启动后台监控线程/服务
		
		executors = Executors.newCachedThreadPool();
		
		// TODO 监控自己的节点, stat/lock/heartbeat/options的变化
		// 如果发现requeststop节点, 则停止
		
		// 监视jobs/triggers/tasks节点的变化
		
		// TODO 启动线程,检查trigger lock是否正常
		
		// TODO 启动线程,检查task是否正常

		// TODO 启动线程,归档陈旧的task到归档目录或别的文件存储目录。
		
	}
	
	private void teardown(){
		
		if(this.provider!=null){
			this.provider.removeScheduler(options.getId());			
		}
		
		// 停止后台线程和服务
		
		// 断开连接
		if(zooKeeper!=null){
			try {
				zooKeeper.close();
			} catch (InterruptedException e) { 
				// ignore
			} finally {
				zooKeeper = null;
			}
		}
		
		// 关闭线程池
		if(executors!=null){ executors.shutdown(); }
	}
	
	private void poll(String triggerId) {
		
		if(logger.isTraceEnabled()){
			logger.trace("Scheduler #{} scanning Trigger #{}", options.getId(), triggerId);
		}
		
		// 判断是否需要触发
		if( ! checkTrigger(triggerId)){
			return;
		}

		String lockPath = options.getTriggersPath()+"/"+triggerId+"/lock";
		
		// 尝试锁定Trigger
//		if(zks.exists(lockPath)){
//			logger.trace("Scheduler #{} scanning Trigger #{} , skiped by trigger lock already exists", 
//					options.getId(), triggerId);
//			return;
//		}
		
		if( ! zks.createEphemeralRecursively(lockPath, instanceId.getBytes())){
			logger.debug("Scheduler #{} scanning Trigger #{} , try to lock trigger failed ", options.getId(), triggerId);
			return;
		}else{
			logger.trace("Scheduler #{} scanning Trigger #{} , try to lock trigger success", options.getId(), triggerId);
		}
		
		// 锁定成功, 创建执行任务 
		try{
			fire(triggerId);
		}finally{
			// 解除锁定
			if( ! zks.delete(lockPath, -1, DeleteStrategy.DeleteAnyway)){
				logger.error("Scheduler #{} scanning Trigger #{} , remove trigger lock failed", 
						options.getId(), triggerId);
				return;
			}else{
				logger.trace("Scheduler #{} scanning Trigger #{} , remove trigger lock success", 
						options.getId(), triggerId);
			}
		}

	}

	private void fire(String triggerId) {
		
		String triggerPath = options.getTriggersPath()+"/"+triggerId;
		// 获取trigger, calendar和job快照
		Trigger trigger = zks.readObject(triggerPath, this.serializer);
		if(trigger==null){
			logger.trace("Scheduler #{} scanning Trigger #{}, trigger not exists after locked it.", options.getId(), triggerId);
			return;
		}else if( !trigger.isEnabled() || ( trigger.getEndTime()>0 && trigger.getEndTime() < System.currentTimeMillis() )){
			logger.trace("Scheduler #{} scanning Trigger #{} , skiped by isEnabled={}, stopTime={}", 
					options.getId(), triggerId, trigger.isEnabled(), trigger.getEndTime());
			return; 
		}
		
		Calendar calendar = null;
		if(trigger.getCalendarId()!=null){
			calendar = zks.readObject("/calendars/"+trigger.getCalendarId(), serializer, null);
			if(calendar==null){
				logger.trace("Scheduler #{} scanning Trigger #{} , skiped by calendar #{} is not exists", 
						options.getId(), triggerId, trigger.getJobId());
				return;
			}
		}

		Job job = zks.readObject(options.getJobsPath()+"/"+trigger.getJobId(), this.serializer);
		if(job==null || !job.isEnabled()){
			logger.trace("Scheduler #{} scanning Trigger #{} , skiped by job #{} is null or disabled", 
					options.getId(), triggerId, trigger.getJobId());
			return;
		}
		
		if(!trigger.isInitialized()){
			if( ! trigger.initialize(calendar)){
				logger.warn("Scheduler #{} scanning Trigger #{} , initialize trigger failed, state={}", 
						options.getId(), triggerId, trigger.getState());
				return;
			}else{
				logger.trace("Scheduler #{} scanning Trigger #{} , initialize trigger success ", options.getId(), triggerId);
			}
		}
		
		long now = System.currentTimeMillis();
		long next = trigger.getNextFireTime();
		if(next > now){ // 还没到下一次执行时间
			if(logger.isTraceEnabled()){
				logger.trace("Scheduler #{} skiped   Trigger #{}, remaining {} MS", options.getId(), triggerId, next - now);
			}
			
			if(minNext==0){
				minNext = next;
			}else{
				minNext = Math.min(minNext, next);						
			}
			return; 
		}else if( trigger.updateAfterMisfire(calendar) ){
//			// 修正后的trigger要及时保存
//			if( null == zks.writeObject(triggerPath, trigger)){
//				logger.warn("Scheduler #{} scanning Trigger #{} , update trigger failed", options.getId(), trigger.getId());
//				return;
//			}else{
//				logger.trace("Scheduler #{} scanning Trigger #{} , update trigger suceess", options.getId(), trigger.getId());
//			}
			next = trigger.getNextFireTime();
		}
		
		if(next<=0 || next>System.currentTimeMillis() || !trigger.isEnabled()){
			if(logger.isTraceEnabled()){
				logger.trace("Scheduler #{} scanning Trigger #{} , skiped by isEnabled={}, fireTime={}", 
						options.getId(), triggerId, trigger.isEnabled(), format(next));
			}
			return;
		}
		
		if(logger.isTraceEnabled()){
			logger.trace("Scheduler #{} scanning Trigger #{}, fire time is {} - {} = delay {} MS", 
					options.getId(), triggerId, format(now), format(next), now-next );
		}
		
		Task task = createTask(trigger,job,calendar);
		if(task!=null){

			// 3. 更新trigger
			trigger = (Trigger) (trigger).clone();	
			trigger.triggered(calendar);
			
			if( null == zks.writeObject(triggerPath, trigger)){
				logger.warn("Scheduler #{} create Task #{}, update trigger failed", options.getId(), task.getId());
				
				// 更新trigger不成功, task丢弃
				if( ! zks.delete(options.getTasksPath()+"/"+task.getId(), -1, DeleteStrategy.DeleteAnyway)){
					logger.error("Scheduler #{} create Task #{}, delete task failed", options.getId(), trigger.getId(), task.getId());
				}
				return;
			}else{
				logger.trace("Scheduler #{} scanning Trigger #{} , update trigger suceess", options.getId(), trigger.getId());
			}

			// TODO trigger完成处理.
			// 根据trigger的设置, 可能删除trigger,或做其他的操作.
			
			
			executors.execute(()->execute(task)); // 放入线程池执行
		}
	}
	
	private boolean checkTrigger(String triggerId){

		String triggerPath = options.getTriggersPath()+"/"+triggerId;
		Trigger trigger = zks.readObject(triggerPath, this.serializer);
		if(trigger==null){
			logger.trace("Scheduler #{} scanning Trigger #{}, can not read trigger data.", options.getId(), triggerId);
			return false;
		}
		
		if( !trigger.isEnabled() || ( trigger.getEndTime()>0 && trigger.getEndTime() < System.currentTimeMillis() )){
			if(logger.isTraceEnabled()){
				logger.trace("Scheduler #{} scanning Trigger #{} , skiped by isEnabled={}, stopTime={}", 
						options.getId(), trigger.getId(), trigger.isEnabled(), trigger.getEndTime());
			}
			return false;
		}
				
		// 计算下一次执行的时间点
		if(trigger.isInitialized()){
			long now = System.currentTimeMillis();
			long next = trigger.getNextFireTime();
			if(next<=0){
				if(logger.isTraceEnabled()){
					logger.trace("Scheduler #{} scanning Trigger #{} , have not valid fire time.", 
							options.getId(), trigger.getId());
				}
				return false;
			}
			
			if(next > now){
				
				if(minNext==0){
					minNext = next;
				}else{
					minNext = Math.min(minNext, next);						
				}
				if(logger.isTraceEnabled()){
					logger.trace("Scheduler #{} skiped   Trigger #{}, remaining {} MS", 
							options.getId(), triggerId, next - now);
				}
				return false; // 还没到下一次执行时间
			}
		}
		
		return true;
	}
	
	private Task createTask(Trigger triggerSnapshot, Job jobSnapshot, Calendar calendarSnapshot){
		
		String triggerId = triggerSnapshot.getId();
		
		// 1. task 
		Task task = new Task();
		task.setId(format2(triggerSnapshot.getNextFireTime())+"-"+triggerId);
		task.setCreator(options.getId());
		task.setCreateTime(System.currentTimeMillis());
		task.setStatus(Task.State.NONE);
		task.setJobSnapshot(jobSnapshot);
		task.setTriggerSnapshot(triggerSnapshot);
		task.setCalendarSnapshot(calendarSnapshot);

		String statefulPath = options.getTriggersPath()+"/"+triggerId+"/stateful.lock";
		if(jobSnapshot.isStateful()){
			if( ! zks.createEphemeral(statefulPath, (instanceId+"#"+task.getId()).getBytes()) ){
				logger.warn("Scheduler #{} create Task #{}, create stateful.lock failed", 
						options.getId(), triggerId, task.getId());
				return null;
			}
		}
		
		// 2. 创建task节点
		String taskPath = options.getTasksPath()+"/"+task.getId();
		if( ! zks.createEphemeralObject(taskPath, task, CreateStrategy.CreateRecursively, CreateStrategy.Ignore)){
			logger.warn("Scheduler #{} create Task #{}, create task failed", 
				options.getId(), triggerId, task.getId());
			return null;
		}
		
		return task;
	}
	
	private void execute(Task task){
		
		String taskPath = options.getTasksPath()+"/"+task.getId();
		
		try{
			
			logger.info("Scheduler #{} execute  Task    #{}           ", options.getId(), task.getId());
			
			task.setStartTime(System.currentTimeMillis());
			task.setStatus(Task.State.NORMAL);
			
			// 更新task
			if( null == zks.writeObject(taskPath, task)){
				logger.warn("Scheduler #{} execute  Task    #{} , update task status #{} to #{} failed", 
					options.getId(), task.getId(), task.getStatus(), Task.State.NORMAL);
			}
			
			invoke(task); // 执行
			
			// 更新task
			task.setStopTime(System.currentTimeMillis());
			task.setStatus(Task.State.COMPLETE);
			
			if( null == zks.writeObject(taskPath, task)){
				logger.warn("Scheduler #{} execute  Task    #{} , update task status #{} to #{} failed", 
						options.getId(), task.getId(), task.getStatus(), Task.State.COMPLETE);
			}
			
		}catch(Exception e){

			logger.error("Scheduler #{} execute  Task    #{} failed, message={}", 
					options.getId(), task.getId(), e.getMessage(), e);
			
			task.setStopTime(System.currentTimeMillis());
			task.setStatus(Task.State.ERROR);
			
			if( null == zks.writeObject(taskPath, task)){
				logger.warn("Scheduler #{} execute  Task    #{} , update task status #{} to #{} failed", 
						options.getId(), task.getId(), task.getStatus(), Task.State.ERROR);
			}
			
		}finally{
			
			try{
				
				logger.trace("Scheduler #{} moving   Task    #{} to history", options.getId(), task.getId());
				
				// 历史分日期存放年
				String historyPath = options.getHistoriesPath()+"/"+task.getId();

				if( null == zks.writeObject(historyPath, task, SetDataStrategy.CreateRecursively )){
					logger.error("Scheduler #{} create   Task    #{}'s history failed", options.getId(), task.getId());
					// TODO 将task信息写入到日志, 因为保存到ZooKeeper失败了
				}
				
				if( ! zks.delete(taskPath, -1)){
					logger.error("Scheduler #{} remove   Task    #{} failed",  options.getId(), task.getId());
					// TODO 删不掉的要有监控线程定期清理
				}
				
			} catch (Exception e) {
				logger.warn("Scheduler #{} save     Task    #{} history failed, message={}", 
						options.getId(), task.getId(), e.getMessage(), e);
			}finally{
				// 删除状态锁
				if( task.getJobSnapshot().isStateful() ){
					String statefulPath = options.getTriggersPath()+"/"+task.getTriggerSnapshot().getId()+"/stateful.lock";
					try {
						if( zks.exists(statefulPath)
							//&& (instanceId+"#"+task.getId()).equals(zks.readString(statefulPath, ""))
						){
							if(zks.delete(statefulPath, -1)){
								logger.error("Scheduler #{} remove   Task    #{} stateful.lock failed",  options.getId(), task.getId());
								// TODO 删不掉的要有监控线程定期清理
							}else{
								logger.error("Scheduler #{} remove   Task    #{} stateful.lock success",  options.getId(), task.getId());
							}
						}
					} catch (Exception e) {
						logger.warn("Scheduler #{} executed Task    #{} , remove trigger lock failed, message={}", 
								options.getId(), task.getId(), e.getMessage(), e);
					}
				}
			}
		}
	}
	
	private void invoke(Task task) throws ClassNotFoundException,
		NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException 
	{
		Job job = task.getJobSnapshot();
		Class<?> clazz = Class.forName(job.getClazz());
		try {
			// 先尝试带map参数的
			Method method = clazz.getMethod(job.getMethod(),Map.class);
			Map<String,Object> params = new HashMap<>();
			params.put("_task", task);
			
			if(Modifier.isStatic(method.getModifiers())){
				method.invoke(null, params);
			}else{
				method.invoke(clazz.newInstance(), params);
			}
		} catch (NoSuchMethodException e) {
			// 再尝试无参数的
			Method method = clazz.getMethod(job.getMethod());
			if(Modifier.isStatic(method.getModifiers())){
				method.invoke(null);
			}else{
				method.invoke(clazz.newInstance());
			}
		}
	}

	///////////////////////////////////////////////////////////////////////////
	// private method
	///////////////////////////////////////////////////////////////////////////

	void updateState(Status status) throws IOException {

		Status old = this.state;
		this.state = status;
		
		if(null == zks.setData(options.getStatusPath(), status.toString().getBytes())){
			logger.warn("Scheduler #{} update state from {} to {} failed", options.getId(), old, status);
		}else{
			logger.info("Scheduler #{} change state from {} to {}", options.getId(), old, status);
		}
		
	}
	
	///////////////////////////////////////////////////////////////////////////
	// getter setter
	///////////////////////////////////////////////////////////////////////////
	
    ///////////////////////////////////////////////////////////////////////////
    // getter and setter
    ///////////////////////////////////////////////////////////////////////////
	
	public String getId() {
		return options.getId();
	}
	
	///////////////////////////////////////////////////////////////////////////
	// tool functions
	///////////////////////////////////////////////////////////////////////////

	private String format(long time){
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
		return sdf.format(new java.util.Date(time));
	}
	
	private String format2(long time){
		SimpleDateFormat sdf = new SimpleDateFormat("yyMMddHHmmssSSS");
		return sdf.format(new java.util.Date(time));
	}
	///////////////////////////////////////////////////////////////////////////
	// inner classes
	///////////////////////////////////////////////////////////////////////////

	public static class Options implements Scheduler.Options {
	
		private static final long serialVersionUID = 1L;
		
		private String id;
		private String name;
		private String description;
		boolean enabled = true;
		
	    private String host = "localhost:2181";
		private String path = "";
	    private int timeout = 30_000;
		private int heartbeat;
	    
	    private String username;
	    private String password;

	    private String serializer="default";
	
		private String jobsPath;
		private String tasksPath;
		private String triggersPath;
		private String calendarsPath;
		private String historiesPath;
		private String schedulersPath;
		
		private String schedulerPath;
		private String statusPath;
		private String optionsPath;
		
	    ///////////////////////////////////////////////////////////////////////////
	    // builder methods
	    ///////////////////////////////////////////////////////////////////////////
	    
	    public static Options from(String uri){
	    	return from(URI.create(uri), Collections.emptyMap());
	    }
	    
	    public static Options from(URI uri){
	    	return from(uri, Collections.emptyMap());
	    }
	
	    @SuppressWarnings("deprecation")
		public static Options from(URI uri, Map<String, ?> env){
	
			Map<String,String> opt = new HashMap<>();
			if(env!=null){
				for (String key : env.keySet()) {
					opt.put(key, env.get(key)!=null ? env.get(key).toString():null );
				}
			}
			
			String auth = uri.getAuthority();
			if(auth!=null && auth.indexOf('@')>0){
				auth = auth.substring(0,auth.indexOf('@'));
				String[] up = auth.split(":");
				if(!up[0].trim().isEmpty()){
					opt.put("username",up[0].trim());
				}
				
				if(up.length==2 && !up[1].trim().isEmpty()){
					opt.put("password", up[1].trim());
				}
			}
			
			List<String> hosts = new ArrayList<>();
			List<Integer> ports = new ArrayList<>();
			if(uri.getQuery()!=null){
		    	String[] params = uri.getQuery().split("&");
		        for (String param : params) {
		            String[] pair = param.split("=");
		            
		            if("host".equals(pair[0])){
		            	hosts.add(pair[1]);
		            }else if("port".equals(pair[0])){
		            	ports.add(Integer.parseInt(pair[1]));
		            }else{
		            	opt.put(pair[0], URLDecoder.decode(pair[1]));
		            }
		        }
	    	}
			
			int port = 2181;
			if(uri.getPort()!=-1){
				port = uri.getPort();
			}
			String host = uri.getHost()+ ":" + port;
			int counter = 0;
			for (String h : hosts) {
				host+=","+h+":"+ (ports.size()>counter?ports.get(counter):port);
				counter++;
			}
			
			opt.put("host", host);
			
			String fullPath = uri.getPath();
			if(fullPath==null || fullPath.isEmpty()){
				throw new IllegalArgumentException("uri is not valid: "+uri);
			}
			
			String[] paths = fullPath.split("/");
			if(paths.length<2 || !paths[0].isEmpty()){
				throw new IllegalArgumentException("uri is not valid: "+uri);
			}
			
			String path="";
			for (int i = 1; i < paths.length-1; i++) {
				if(!paths[i].isEmpty()){
					path+="/"+paths[i];					
				}
			}

			opt.put("id", paths[paths.length-1]);
			opt.put("path", path);
			
			return from(opt);
	    }
	    
	    public static Options from(Map<String, String> env){
	    	Options op = new Options();

	    	op.id             = env.getOrDefault("id", op.id);
	    	op.name           = env.getOrDefault("name", op.id);
	    	op.description    = env.getOrDefault("description", "");
	    	op.enabled        = Boolean.valueOf(env.getOrDefault("enable", "true"));
	    	
	    	op.host           = env.getOrDefault("host", op.host);
	    	op.path           = env.getOrDefault("path", op.path);
	    	op.timeout        = Integer.parseInt(env.getOrDefault("timeout", ""+op.timeout));
	    	op.heartbeat      = Integer.parseInt(env.getOrDefault("heartbeat", ""+op.heartbeat));

	    	op.username       = env.getOrDefault("username", op.username);
	    	op.password       = env.getOrDefault("password", op.password);
	    	
	    	op.serializer     = env.getOrDefault("serializer", op.serializer);
	
			op.jobsPath       = "/jobs";
			op.tasksPath      = "/tasks";
			op.triggersPath   = "/triggers";
			op.calendarsPath  = "/calendars";
			op.historiesPath  = "/histories";
			
			op.schedulersPath = "/schedulers";
			op.schedulerPath  = op.schedulersPath + "/" + op.id;
			op.statusPath     = op.schedulerPath  + "/status";
			op.optionsPath    = op.schedulerPath  + "/options";
			
			if(Boolean.valueOf(env.getOrDefault("verbose", "false"))){
				System.out.println(op.toPrettyString());
			}
			
	    	return op;
	    }
	
	    ///////////////////////////////////////////////////////////////////////////
	    //  methods
	    ///////////////////////////////////////////////////////////////////////////
	    
	    public boolean isValid(){
	    	return validate().isEmpty();
	    }
	
	    public List<String> validate(){
	    	// TODO 配置项合法性检查
	    	return Collections.emptyList();
	    }
	    
//	    public String getPrefixPath(){
//	    	return url.indexOf('/')==-1? null : url.substring(url.indexOf('/'));
//	    }
	    
		public URI toURI() {
			return URI.create(getFullPath(null));
		}
	
	    @SuppressWarnings("deprecation")
		public String getFullPath(String path) {
			if(path!=null && !path.isEmpty() && path.charAt(0)!='/'){
				throw new IllegalArgumentException("argument <path> must start with '/' or is null");
			}
	
			StringBuilder buf = new StringBuilder("zookeeper://");
			
			if(this.username!=null){
				buf.append(this.username);
				buf.append("@");
			}
			
			String[] hosts = {};
//			String paths="";
//			int index = url.indexOf('/');
//			if(index>-1){
//				hosts = url.substring(0,index).split(",");
//				paths = url.substring(index);
//			}else{
				hosts = this.host.split(",");
//			}
//			
//			if(path!=null){
//				paths+=path;
//			}
			
			buf.append(hosts[0]).append(this.path).append("/").append(this.id);
			
			buf.append("?").append("timeout=").append(this.timeout);
			if(this.password!=null){
				buf.append("&password=").append(URLEncoder.encode(this.password));
			}
			if(this.serializer!=null && ! "default".equals(this.serializer)){
				buf.append("&serializer=").append(URLEncoder.encode(this.serializer));
			}
			
			for (int i = 1; i < hosts.length; i++) {
				String[] hp = hosts[i].split(":");
				buf.append("&").append("host=").append(hp[0]);
				buf.append("&").append("port=").append(hp[1]);
			}
			
			return buf.toString();
		}
		
		@Override
		public String toString() {
			return getFullPath(null);
		}
	    
		public String toPrettyString(){
			String str="ZooKeeperScheduler Options: \n";

			str+="Scheduler ID   : "+this.id +"\n";
			str+="Scheduler Name : "+this.name +"\n";
			str+="Is Enabled     : "+ this.enabled +"\n";
			
			str+="Scheme         : "+ "zookeeper" +"\n";
			str+="Connection URL : "+this.host+this.path +"\n";
			str+="Timeout        : "+this.timeout +"\n";
			str+="Heartbeat      : "+this.heartbeat +"\n";
			
			str+="UserName       : "+(this.username==null?"<NULL>":this.username) +"\n";
			str+="Password       : "+(this.password==null?"<NULL>":"******") +"\n";
			str+="Serializer     : "+this.serializer +"\n";
	
			str+="Job       Path : "+this.jobsPath +"\n";
			str+="Task      Path : "+this.tasksPath +"\n";
			str+="Trigger   Path : "+this.triggersPath +"\n";
			str+="Calendar  Path : "+this.calendarsPath +"\n";
			str+="History   Path : "+this.historiesPath +"\n";
			
			str+="Scheduler Path : "+this.schedulerPath +"\n";
			str+="Status    Path : "+this.statusPath +"\n";
			str+="Options   Path : "+this.optionsPath +"\n";
			
			str+="Description    : "+this.description +"\n";
			
			return str;
		}
		
	    ///////////////////////////////////////////////////////////////////////////
	    // getter and setter
	    ///////////////////////////////////////////////////////////////////////////
	    
		public String getUrl() {
			return host+path;
		}
//		public void setUrl(String url) {
//			this.url = url;
//		}
		
		public int getTimeout() {
			return timeout;
		}
		public void setTimeout(int timeout) {
			this.timeout = timeout;
		}
	
		public String getUserName() {
			return username;
		}
	
		public void setUserName(String username) {
			this.username = username;
		}
	
		public String getPassword() {
			return password;
		}
	
		public void setPassword(String password) {
			this.password = password;
		}
	
		public String getId() {
			return id;
		}

		public String getHost() {
			return host;
		}
		public String getPath() {
			return path;
		}
	
		public int getHeartbeat() {
			return heartbeat;
		}
		public String getName() {
			return name;
		}
		public void setName(String name) {
			this.name = name;
		}
		public String getDescription() {
			return description;
		}
		public void setDescription(String description) {
			this.description = description;
		}
	
		public String getJobsPath() {
			return jobsPath;
		}
	
		public String getTasksPath() {
			return tasksPath;
		}
	
		public String getTriggersPath() {
			return triggersPath;
		}
	
		public String getHistoriesPath() {
			return historiesPath;
		}
	
		public String getSchedulersPath() {
			return schedulersPath;
		}
	
		public String getSchedulerPath() {
			return schedulerPath;
		}
	
		public String getStatusPath() {
			return statusPath;
		}
	
		public String getOptionsPath() {
			return optionsPath;
		}
	
		public boolean isEnabled() {
			return enabled;
		}
	
		public void setEnabled(boolean enabled) {
			this.enabled = enabled;
		}
	
		public String getSerializer() {
			return serializer;
		}

		public void setSerializer(String serializer) {
			this.serializer = serializer;
		}

		public String getCalendarsPath() {
			return calendarsPath;
		}

		public Options merge(Options other) {
			if(this==other){
				return this;
			}
			
			
			
			// TODO 未实现...
			return this;
		}	
	
	}

}
